package com.gitee.xhs.windowState;

import com.gitee.xhs.EnvironmentConfig;
import com.gitee.xhs.FlinkEnvironmentFactory;
import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

import static com.gitee.xhs.windowState.WindowStateConstant.TUMBLING_EVENT_TIME_WINDOWS_ONE_MINUTE;
import static com.gitee.xhs.windowState.WindowStateConstant.WINDOW_STATE_UID;


/**
 * @author BigHead
 * @date 2022/7/7 17:15
 * @description WindowStateMain
 */
public class WindowStateMain {

    public static void main(String[] args) throws Exception {
        EnvironmentConfig environmentConfiguration = new EnvironmentConfig(null);
        environmentConfiguration.checkpoint(
                null,
                null,
                WindowStateConstant.WINDOW_STATE_PATH,
                null
        );

        StreamExecutionEnvironment streamExecutionEnvironment = FlinkEnvironmentFactory.simpleEnv(environmentConfiguration);
        streamExecutionEnvironment
                .addSource(new TransactionUnBoundFiveBatchSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<Transaction>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                                .withTimestampAssigner(
                                        (transaction, timestamp) -> transaction.getTimestamp()
                                )
                )
                .keyBy(Transaction::getAccountId)
                .window(TUMBLING_EVENT_TIME_WINDOWS_ONE_MINUTE)
                .aggregate(new TransactionOneMinuteCounter()).uid(WINDOW_STATE_UID)
                .print();

        streamExecutionEnvironment.execute("WindowStateMain");
    }

}
